package com.example.sparkkafka

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, TaskContext}

object OffsetWithRedis {
  def main(args: Array[String]): Unit = {
    // 初始化
//    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getCanonicalName)
    val ssc = new StreamingContext(conf, Seconds(1))
    // 定义kafka相关参数
    val groupId: String = "mygroup2"
    val topics: Array[String] = Array("topic1")
    val kafkaParams: Map[String, Object] = getKafkaConsumerParameters(groupId)
    // 从redis中获取根据主题和消费组获取offsets
    val offsets: Map[TopicPartition, Long] = OffsetsRedisUtils.getOffsetFromRedis(topics, groupId)
    println(offsets)
    
    // 创建DStream
    val dstream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream(
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsets)
      )

    // DStream转换&输出
    println(dstream.hashCode())
    dstream.foreachRDD { (rdd, time) =>
      println(s"count: ${rdd.count}")
      if (!rdd.isEmpty()) {
        // 处理消息
        println(s"*********** rdd.count = ${rdd.count()}; time = $time *************")
        // 将offsets保存到redis,下次消费不会重复消费数据
        val offsetRanges: Array[OffsetRange] =
          rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        OffsetsRedisUtils.saveOffsetsToRedis(offsetRanges, groupId)
      }

    }
    // 启动作业
    ssc.start()
    ssc.awaitTermination()


  }

  def getKafkaConsumerParameters(groupId: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092,node2:9092,node3:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
//      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
    )
  }
}
